package com.gitee.xmhzzz.rocketmq.quick.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.*;
import org.apache.rocketmq.client.apis.producer.Producer;

import java.util.Collections;

/**
 * @ClassName DefaultMqProvider
 * @Description
 * @Author wzq
 * @Date 2024/12/9 11:15
 * @Version 1.0
 */
@Slf4j
public class DefaultMqProvider {

    private final static String endpoints = "127.0.0.1:18081";

    public Producer createProducer(String topic) {
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        ClientConfiguration configuration = builder.build();

        try {
            return provider.newProducerBuilder()
                    .setTopics(topic)
                    .setClientConfiguration(configuration)
                    .build();
        }catch (Exception e){
            log.error("创建生产者失败 topic[{}] msg[{}]",topic,e.getMessage(),e);
        }

        return null;
    }


    public void createPushConsumer(String consumerGroup, String topic, MessageListener listener) {
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        ClientConfiguration clientConfiguration  = builder.build();
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        try {
             provider.newPushConsumerBuilder()
                    .setClientConfiguration(clientConfiguration)
                    //设置消费者分组。
                    .setConsumerGroup(consumerGroup)
                    //设置预绑定的订阅关系。
                    .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                    //设置消费监听器。
                    .setMessageListener(listener)
                    .build();
            Thread.sleep(Long.MAX_VALUE);
        }catch (Exception e){
            log.error("创建消费者失败 topic[{}] msg[{}]",topic,e.getMessage(),e);
        }

    }

    //messageView -> {
    //                         //处理消息并返回消费结果。
    //                         log.info("Consume message={}", messageView);
    //                         return ConsumeResult.SUCCESS;
    //                     }

}
